package com.runjian.rundo.iot.agent.business.publisher.impl;

import com.alibaba.fastjson2.JSON;
import com.runjian.rundo.iot.agent.business.config.BaseConfigurationProperties;
import com.runjian.rundo.iot.agent.business.kafka.KafkaSender;
import com.runjian.rundo.iot.agent.business.publisher.MessagePublisher;
import com.runjian.rundo.iot.agent.common.message.DecodedMessage;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 *
 * @author CaoQingHua
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaMessagePublisher implements MessagePublisher {

    private final BaseConfigurationProperties businessTopicConfig;

    private final KafkaSender kafkaSender;

    @Override
    public void sendDeviceMessage(DecodedMessage decodedMessage) {
        String jsonContent = JSON.toJSONString(decodedMessage);
        if (log.isDebugEnabled()) {
            log.debug("发送设备实时数据消息 Topic名称:{}, 数据内容: {}", businessTopicConfig.getMeasurementsUpTopic(), jsonContent);;
        }
        kafkaSender.send(businessTopicConfig.getMeasurementsUpTopic(), jsonContent);
    }

    @Override
    public void sendVppMessage(DecodedMessage vppMessage) {
        String jsonContent = JSON.toJSONString(vppMessage);
        if (log.isDebugEnabled()) {
            log.debug("发送虚拟电厂总指令数据消息 Topic名称:{}, 数据内容: {}", businessTopicConfig.getControlUpTopic(), jsonContent);;
        }
        kafkaSender.send(businessTopicConfig.getControlUpTopic(), jsonContent);
    }

    @Override
    public void send(String topic, Object obj) {
        String jsonContent = JSON.toJSONString(obj);
        if (log.isDebugEnabled()) {
            log.debug("发送数据消息 Topic名称:{}, 数据内容: {}", topic, jsonContent);;
        }
        kafkaSender.send(topic, jsonContent);
    }

}
